package top.codedance.iotp.client.plugins.async.mqtt;

import cd.io.netty.channel.ChannelHandlerContext;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import top.codedance.iotp.client.ClientApplication;
import top.codedance.iotp.common.executer.AsyncAbstractExecuter;
import top.codedance.iotp.common.entity.CommonEntity;
import org.ini4j.Ini;
import org.ini4j.Wini;

import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MqttExecuterImpl extends AsyncAbstractExecuter {
    private static ExecutorService mqttPools = Executors.newCachedThreadPool();
    private static Map<Integer, MqttService> mqtts = new HashMap<>();
    public static volatile Map<String, Set<String>> tasks = new HashMap<>();

    @Override
    public void exec(ChannelHandlerContext channelHandlerContext, CommonEntity commonEntity, String taskId) {
        super.exec(channelHandlerContext, commonEntity, taskId);
        mqttPools.submit(() -> {
            JSONObject requestBodyJSONObject = getRequestBodyJSONObject();
            Integer instance = requestBodyJSONObject.getInteger("instance");
            tryCreateConnect(this, instance);
            if(requestBodyJSONObject.containsKey("body")){
                synchronized (MqttExecuterImpl.class) {
                    JSONArray body = requestBodyJSONObject.getJSONArray("body");
                    String topic;
                    for (int i = 0; i < body.size(); i++) {
                        topic = body.getString(i);
                        try{
                            JSONObject writeObject = JSON.parseObject(topic);
                            MqttService mqttService = mqtts.get(instance);
                            mqttService.send(writeObject.getString("topic"), writeObject.getJSONObject("message"));
                        }catch (Exception ex){
                            if (!tasks.containsKey(topic)) {
                                tasks.put(topic, new LinkedHashSet<>());
                            }
                            tasks.get(topic).add(getClientId());
                        }
                    }
                }
            }
            if(requestBodyJSONObject.containsKey("release") && requestBodyJSONObject.getBoolean("release")){
                synchronized (MqttExecuterImpl.class) {
                    Set<Map.Entry<String, Set<String>>> entries = tasks.entrySet();
                    for (Map.Entry<String, Set<String>> entry : entries) {
                        try {
                            entry.getValue().remove(getClientId());
                        } catch (Exception e) {
                        }
                    }
                }
            }
        });
    }

    private synchronized void tryCreateConnect(AsyncAbstractExecuter executer, Integer instance){
        if(!mqtts.containsKey(instance)) {
            MqttServerParam mqttServerParam = new MqttServerParam();
            Wini ini = getIniConfig();
            if (ini != null && ini.containsKey("mqtt")) {
                Ini.Section section = ini.get("mqtt");
                if (section.containsKey("ip." + instance)) {
                    mqttServerParam.setIp(section.get("ip." + instance));
                }
                if (section.containsKey("port." + instance)) {
                    mqttServerParam.setPort(Integer.parseInt(section.get("port." + instance)));
                }
                if (section.containsKey("username." + instance)) {
                    mqttServerParam.setUsername(section.get("username." + instance));
                }
                if (section.containsKey("password." + instance)) {
                    mqttServerParam.setPassword(section.get("password." + instance));
                }
                if (section.containsKey("topic." + instance)) {
                    mqttServerParam.setTopic(section.get("topic." + instance).split("\\|"));
                }
                if (section.containsKey("qos." + instance)) {
                    String[] split = section.get("qos." + instance).split("\\|");
                    int[] qos = new int[split.length];
                    for (int i = 0;i<split.length;i++){
                        qos[i] = Integer.valueOf(split[i]).intValue();
                    }
                    mqttServerParam.setQos(qos);
                }
            }
            MqttService mqttService = new MqttService(executer, mqttServerParam);
            try{
                mqttService.createConnect();
                mqtts.put(instance, mqttService);
            }catch (Exception e){}
        }
    }
}
